package edu.wisc.hevea.messaging;

import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;

import edu.wisc.hevea.broker.MessageBroker;
import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.exception.InvalidTopicException;
import edu.wisc.hevea.exception.ReplicationException;
import edu.wisc.hevea.hsql.Message;
import edu.wisc.hevea.messaging.Messaging.GetMessageRequest;
import edu.wisc.hevea.messaging.Messaging.GetMessageResponse;
import edu.wisc.hevea.messaging.Messaging.PutMessageRequest;
import edu.wisc.hevea.messaging.Messaging.PutMessageResponse;
import edu.wisc.hevea.messaging.Messaging.PutQueueLengthRequest;
import edu.wisc.hevea.messaging.Messaging.PutQueueLengthResponse;
import edu.wisc.hevea.messaging.Messaging.ReconcileMessageResponse;
import edu.wisc.hevea.messaging.Messaging.ReconcileMessagesRequest;
import edu.wisc.hevea.messaging.Messaging.ReplicateMessageRequest;
import edu.wisc.hevea.messaging.Messaging.ReplicateMessageResponse;
import edu.wisc.hevea.topic.TopicManager;

public class MessagingServiceImpl implements Messaging.MessagingService.Interface {

	private static final Log logger = LogFactory.getLog(MessagingServiceImpl.class);

	private TopicManager topicManager;
	private Map<String,Integer> dcToLastStartedPort;
	private MessageBroker messageBroker;
	
	public MessagingServiceImpl(TopicManager topicManager,MessageBroker messageBroker) {
		dcToLastStartedPort = new HashMap<String, Integer>();
		dcToLastStartedPort.put(ApplicationConstants.VIRGINIA_DC, 10005);
		dcToLastStartedPort.put(ApplicationConstants.IAD_DC, 20005);
		dcToLastStartedPort.put(ApplicationConstants.DUBLIN_DC, 30005);
		dcToLastStartedPort.put(ApplicationConstants.HONGKONG_DC, 40005);
		this.topicManager = topicManager;
		this.messageBroker = messageBroker;
	}

	@Override
	public void putMessage(RpcController controller, PutMessageRequest request,
			RpcCallback<PutMessageResponse> done) {

		String topic = request.getTopic();
		String content = request.getContent();

		logger.info("Received Message :- Topic: " + topic + " Content: "
				+ content);
		PutMessageResponse response = null;
		try {
			Message m = topicManager.handleMessage(topic, content);
			response = PutMessageResponse.newBuilder()
					.setIsSuccess(true).setMsgId(m.getMsgId()).build();
		} catch (InvalidTopicException e) {
			response = PutMessageResponse.newBuilder()
					.setIsSuccess(false).setMsgId("ERROR").build();
		} catch (SQLException e) {
			response = PutMessageResponse.newBuilder()
					.setIsSuccess(false).setMsgId("ERROR").build();			
		} catch (ReplicationException e) {
			response = PutMessageResponse.newBuilder()
					.setIsSuccess(false).setMsgId("ERROR").build();			
		}
		
		done.run(response);
	}

	@Override
	public void getMessage(RpcController controller, GetMessageRequest request,
			RpcCallback<GetMessageResponse> done) {

		String topic = request.getTopic();
		
		logger.info("Got getMessageRequest for topic : " + topic);
		GetMessageResponse response = null;
		try {
			Message m = topicManager.handleGetMessage(topic);
			response = GetMessageResponse.newBuilder()
					.setContent(m.getContent()).setMsgId(m.getMsgId()).setIsSuccess(true).build();
			
			topicManager.handleMessageConsumed(m);
			
		} catch (InvalidTopicException e) {
			response = GetMessageResponse.newBuilder()
					.setIsSuccess(false).build();
		} catch (SQLException e) {
			response = GetMessageResponse.newBuilder()
					.setIsSuccess(false).build();
		}
		
		done.run(response);
		
	}

	@Override
	public void replicateMessage(RpcController controller,
			ReplicateMessageRequest request,
			RpcCallback<ReplicateMessageResponse> done) {
		
		Message m = new Message();
		m.setMsgId(request.getMsgId());
		m.setTopic(request.getTopic());
		m.setContent(request.getContent());
		m.setStatus("PENDING");
		
		ReplicateMessageResponse response = null;
		
		try {
			topicManager.handleMessageReplication(m);
			response = ReplicateMessageResponse.newBuilder().setIsSuccess(true).build();
		} catch (InvalidTopicException e) {
			e.printStackTrace();
			response = ReplicateMessageResponse.newBuilder().setIsSuccess(false).build();
		} catch (SQLException e) {
			response = ReplicateMessageResponse.newBuilder().setIsSuccess(false).build();
			e.printStackTrace();			
		}
		
		done.run(response);
	}

	@Override
	public void reconcileMessages(RpcController controller,
			ReconcileMessagesRequest request,
			RpcCallback<ReconcileMessageResponse> done) {

		List<String> msgIds = request.getMsgIdList();
		ReconcileMessageResponse response = null;
		
		if(msgIds.size() > 0) {
			try {
				topicManager.handleReplicatedMessagesConsumed(msgIds);
				response = ReconcileMessageResponse
						.newBuilder().setIsSuccess(true).build();
			} catch (InvalidTopicException e) {
				response = ReconcileMessageResponse
						.newBuilder().setIsSuccess(false).build();
				e.printStackTrace();
			} catch (SQLException e) {
				response = ReconcileMessageResponse
						.newBuilder().setIsSuccess(false).build();
				e.printStackTrace();
			}			
		} else {
			response = ReconcileMessageResponse
					.newBuilder().setIsSuccess(true).build();
		}
		
		done.run(response);
	}

	@Override
	public void putQueueLength(RpcController controller,
			PutQueueLengthRequest request,
			RpcCallback<PutQueueLengthResponse> done) {
		PutQueueLengthResponse response = PutQueueLengthResponse.newBuilder()
				.setIsSuccess(true).build();
		
		String brokerServerName = request.getHostName();
		int queueLength = request.getQueueLength();
		messageBroker.getAutoScalingThread().getBrokerQueueLengthMap()
				.put(brokerServerName, queueLength);
		
		done.run(response);
	}

}
